package com.tech.common

import org.apache.kudu.Schema
import org.apache.kudu.client.KuduClient.KuduClientBuilder
import org.apache.kudu.client._

import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

object KuduUtil {

  var kuduMaster = ""
  println("加载kudu")
  println(kuduMaster)
  //被transient修饰的变量不参与序列化和反序列化。
  @transient lazy val kuduClient: KuduClient = new KuduClientBuilder(kuduMaster).build()
  @transient lazy val session: KuduSession = kuduClient.newSession()
  //事件表
    @transient lazy val retaileventTable: KuduTable = kuduClient.openTable("retailevent")
  //用户表
    @transient lazy val customerTable: KuduTable = kuduClient.openTable("customer")
  //offset表
  @transient lazy val offsetTable: KuduTable = kuduClient.openTable("offset_test")
//    @transient lazy val offsetTable: KuduTable = kuduClient.openTable("offset")
  //规则表
  @transient lazy val ruleTable: KuduTable = kuduClient.openTable("customer_properties")


  /**
   * 获取表schema
   *
   * @param tableName 表名
   * @return Map(字段名 -> 字段类型)
   */
  def getTableSchema(tableName: String): mutable.HashMap[String, String] = {

    val table = tableName match {
      case "retailevent" =>
        retaileventTable
      case "customer" =>
        customerTable
    }

    val schema = table.getSchema

    val iterator = schema.getColumns.iterator()

    val map = mutable.HashMap[String, String]()

    while (iterator.hasNext) {

      val columnSchema = iterator.next()
      map.put(columnSchema.getName, columnSchema.getType.getName)

    }

    map

  }

  def getRule(): mutable.HashMap[String, mutable.HashMap[String, String]] ={

    // 设置查询条件
    val schema = ruleTable.getSchema
    val projectNotNull = KuduPredicate.newIsNotNullPredicate(schema.getColumn("project"))
    val eventKeyNotNull = KuduPredicate.newIsNotNullPredicate(schema.getColumn("event_key"))
    val updateModeNotNull = KuduPredicate.newIsNotNullPredicate(schema.getColumn("update_mode"))
    val builder = kuduClient.newScannerBuilder(ruleTable)
      .addPredicate(projectNotNull)
      .addPredicate(eventKeyNotNull)
      .addPredicate(updateModeNotNull)

    val scanner = builder.build()

    val map = mutable.HashMap[String, mutable.HashMap[String, String]]()

    while (scanner.hasMoreRows){
      val iterator = scanner.nextRows()
      while (iterator.hasNext){
        val result = iterator.next()
        val project = result.getString("project")
        val eventKey = result.getString("event_key")
        val projectEventKey = project + "|" + eventKey
        val attributeKey = result.getString("attribute_key")
        val updateMode = result.getString("update_mode")
        if (project != "" && eventKey != ""){
          if (map.contains(eventKey)){
            map(projectEventKey).put(attributeKey,updateMode)
          }else{
            val tmpMap = mutable.HashMap[String, String]()
            tmpMap.put(attributeKey,updateMode)
            map.put(projectEventKey,tmpMap)
          }
        }
      }
    }

    map

  }

  /**
   * 获取上次消费截止的offset+1
   * @return Map(topic-partition -> offset)
   */
  def getOffset(): mutable.HashMap[String,Long] = {

    val builder = kuduClient.newScannerBuilder(offsetTable)

    val scanner = builder.build()

    val map = new mutable.HashMap[String, Long]()

    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        val topic = result.getString("topic")
        val partition = result.getInt("partition")
        val offset = result.getLong("offset")
        map.put(topic + "-" + partition, offset)
      }
    }

    map
  }

  /**
   * 更新消费到的offset
   */
  def upsertOffset(topic:String,partition:Int,offset:Long): Unit ={

    val upsert = offsetTable.newUpsert()
    val row = upsert.getRow

    row.addString("topic",topic)
    row.addInt("partition",partition)
    row.addLong("offset",offset + 1)

    session.apply(upsert)
    session.flush()

  }

  /**
   * 插入数据到kudu
   *
   * @param tableName 表名
   * @param map       插入的map
   * @param schemaMap 表的schema
   */
  def insertRow(tableName: String, map: mutable.HashMap[String, Any], schemaMap: mutable.HashMap[String, String]): Unit = {

    val table = tableName match {
      case "retailevent" =>
        retaileventTable
      case "customer" =>
        customerTable
    }

    val upsert = table.newUpsert()
    val row = upsert.getRow

    map.foreach(
      i => {
        if (schemaMap.contains(i._1)) {

          schemaMap(i._1) match {
            case "string" =>
              row.addString(i._1, i._2.toString)
            case "unixtime_micros" =>
              row.addLong(i._1, i._2.toString.toLong * 1000)
            case "double" =>
              row.addDouble(i._1, i._2.toString.toDouble)
            case "int32" =>
              row.addInt(i._1, i._2.toString.toInt)
            case _ =>
              print("异常数据：")
              println(i._2.toString)
          }
        }
      }
    )

    session.apply(upsert)
    session.flush()


  }

  /**
   * 更新用户表信息
   *
   * @param customer_id 用户id
   * @param create_time 创建时间
   * @param map         更新的map（字段名 -> 字段值）
   */
  def upsertCustomer(project: String, customer_id: String, create_time: Long, map: mutable.HashMap[String, Any],customerSchemaMap: mutable.HashMap[String, String]): Unit = {

    val nodeUpsert = customerTable.newUpsert()
    nodeUpsert.getRow.addString("project", project)
    nodeUpsert.getRow.addString("customer_id", customer_id)
    nodeUpsert.getRow.addLong("created_time", create_time * 1000)

    println(map)

    map.foreach(
      t => {
        customerSchemaMap(t._1) match {
          case "string" =>
            nodeUpsert.getRow.addString(t._1, t._2.asInstanceOf[String])
          case "unixtime_micros" =>
            nodeUpsert.getRow.addLong(t._1, t._2.toString.toLong * 1000)
          case "double" =>
            nodeUpsert.getRow.addDouble(t._1, t._2.toString.toDouble)
          case "int32" =>
            nodeUpsert.getRow.addDouble(t._1, t._2.toString.toInt)
          case _ =>
            print("异常数据：")
            println(t._2.toString)
        }
      }
    )

    session.apply(nodeUpsert)
    session.flush()

  }

  /**
   * 查询用户详细信息（指定字段）
   *
   * @param customer_id 用户id
   * @param array       指定字段array
   * @return map（指定字段名 -> 字段值）
   */
  def getCustomerDetail(project: String, customer_id: String, array: ArrayBuffer[String]): mutable.HashMap[String, Any] = {

    // 设置查询条件
    val schema = customerTable.getSchema
    val projectId = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, project)
    val customerId = KuduPredicate.newComparisonPredicate(schema.getColumn("customer_id"), KuduPredicate.ComparisonOp.EQUAL, customer_id)

    // 执行查询操作
    val builder = kuduClient.newScannerBuilder(customerTable).setProjectedColumnNames(array.asJava)
    builder.addPredicate(projectId)
    builder.addPredicate(customerId)

    val scanner = builder.build()

    val map = mutable.HashMap[String, Any]()
    //    println(println("开始循环 ==> " + event_key + "    " + customer_id + "     " + System.currentTimeMillis()))
    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        array.foreach(
          i => {
            if (!result.isNull(i)) {
              map.put(i, schema.getColumn(i).getType.getName match {
                case "double" => result.getDouble(i)
                case "string" => result.getString(i)
                case "unixtime_micros" => result.getLong(i) / 1000
                case "int32" => result.getInt(i)
                case _ => null
              })
            }
          }
        )
      }
    }
    //    println(println("结束循环 ==> " + event_key + "    " + customer_id + "     " + System.currentTimeMillis()))

    scanner.close()

    map

  }

  /**
   * 获取用户详细信息
   *
   * @param customer_id 用户id
   * @return RowResult
   */
  def getCustomerDetail(project: String, customer_id: String): RowResult = {

    // 设置查询条件
    val schema = customerTable.getSchema
    //    val project = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, "")
    val projectId = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, project)
    val customerId = KuduPredicate.newComparisonPredicate(schema.getColumn("customer_id"), KuduPredicate.ComparisonOp.EQUAL, customer_id)

    // 执行查询操作
    val scanner = kuduClient.newScannerBuilder(customerTable)
      .addPredicate(projectId)
      .addPredicate(customerId)
      .build()
    var result: RowResult = null
    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        result = iterator.next()
      }
    }
    scanner.close()

    result

  }

}
